package com.xfsg.core.kafka;

import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.springframework.aop.framework.ProxyFactoryBean;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;
import org.springframework.kafka.listener.MessageListener;
import org.springframework.kafka.listener.MessageListenerContainer;

/**
 * @author ZhangYi
 * @version 1.0.0
 * @Description:
 * @date 2022/3/24
 */
@Slf4j
@Aspect
public class TracingKafkaAspect {

    @Pointcut("execution(public * org.springframework.kafka.config.KafkaListenerContainerFactory.createListenerContainer(..))")
    private void anyCreateListenerContainer() {
    }

    @Pointcut("execution(public * org.springframework.kafka.config.KafkaListenerContainerFactory.createContainer(..))")
    private void anyCreateContainer() {
    }

    @Around("anyCreateListenerContainer() || anyCreateContainer()")
    public Object wrapListenerContainerCreation(ProceedingJoinPoint pjp) throws Throwable {
        MessageListenerContainer listener = (MessageListenerContainer) pjp.proceed();
        if (listener instanceof AbstractMessageListenerContainer){
            AbstractMessageListenerContainer container = (AbstractMessageListenerContainer) listener;
            Object someMessageListener = container.getContainerProperties().getMessageListener();
            if (someMessageListener == null) {
                if (log.isDebugEnabled()) {
                    log.debug("No message listener to wrap. Proceeding");
                }
            }
            else if (someMessageListener instanceof MessageListener) {
                container.setupMessageListener(createProxy(someMessageListener));
            }
            else {
                if (log.isDebugEnabled()) {
                    log.debug("ATM we don't support Batch message listeners");
                }
            }
        }else {
            if (log.isDebugEnabled()){
                log.debug("Can't wrap this listener. Proceeding");
            }
        }
        return listener;
    }

    Object createProxy(Object bean) {
        ProxyFactoryBean factory = new ProxyFactoryBean();
        factory.setProxyTargetClass(true);
        factory.addAdvice(new TracingMessageListenerMethodInterceptor());
        factory.setTarget(bean);
        return factory.getObject();
    }

}
